package cn.linjianhui.flink.sample.function;

import cn.linjianhui.flink.sample.model.OrderAccumulator;
import cn.linjianhui.flink.sample.model.OrderOutputResult;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

public class OrderKeyedProcessFunction extends KeyedProcessFunction<Tuple, OrderAccumulator, OrderOutputResult> {
    private MapState<String, OrderAccumulator> state;

    @Override
    public void open(Configuration parameters) throws Exception {
        state = this.getRuntimeContext().getMapState(new MapStateDescriptor<>(
                "site-pay-total-state", String.class, OrderAccumulator.class));
    }

    @Override
    public void close() throws Exception {
        state.clear();
    }

    @Override
    public void processElement(OrderAccumulator accumulator, Context context, Collector<OrderOutputResult> collector) throws Exception {
        String key = accumulator.getOrderSource();
        OrderAccumulator cachedValue = state.get(key);

        //TODO: 待解决Bug 只有一个下单，判断不成立
        if(cachedValue == null || !accumulator.getTotalPayOrder().equals(cachedValue.getTotalPayOrder())
                || !accumulator.getCancelOrderNum().equals(cachedValue.getCancelOrderNum())) {
            OrderOutputResult statistics = new OrderOutputResult(Integer.valueOf(accumulator.getOrderSource()),
                    accumulator.getTotalPayAmount().doubleValue(), accumulator.getTotalPayOrder(),
                    accumulator.getCancelPayAmount().doubleValue(), accumulator.getCancelOrderNum());
            collector.collect(statistics);
            state.put(accumulator.getOrderSource(), accumulator);
        }
    }
}
